package com.atguigu.tingshu.album.service.impl;

import cn.hutool.core.bean.BeanUtil;
import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.util.IdUtil;
import cn.hutool.core.util.RandomUtil;
import com.atguigu.tingshu.album.mapper.AlbumAttributeValueMapper;
import com.atguigu.tingshu.album.mapper.AlbumInfoMapper;
import com.atguigu.tingshu.album.mapper.AlbumStatMapper;
import com.atguigu.tingshu.album.mapper.TrackInfoMapper;
import com.atguigu.tingshu.album.service.AlbumInfoService;
import com.atguigu.tingshu.common.cache.GuiGuCache;
import com.atguigu.tingshu.common.constant.KafkaConstant;
import com.atguigu.tingshu.common.constant.RedisConstant;
import com.atguigu.tingshu.common.constant.SystemConstant;
import com.atguigu.tingshu.common.execption.GuiguException;
import com.atguigu.tingshu.common.service.KafkaService;
import com.atguigu.tingshu.model.album.AlbumAttributeValue;
import com.atguigu.tingshu.model.album.AlbumInfo;
import com.atguigu.tingshu.model.album.AlbumStat;
import com.atguigu.tingshu.model.album.TrackInfo;
import com.atguigu.tingshu.query.album.AlbumInfoQuery;
import com.atguigu.tingshu.vo.album.AlbumAttributeValueVo;
import com.atguigu.tingshu.vo.album.AlbumInfoVo;
import com.atguigu.tingshu.vo.album.AlbumListVo;
import com.atguigu.tingshu.vo.album.AlbumStatVo;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.util.Arrays;
import java.util.List;
import java.util.concurrent.TimeUnit;

@Slf4j
@Service
@SuppressWarnings({"all"})
public class AlbumInfoServiceImpl extends ServiceImpl<AlbumInfoMapper, AlbumInfo> implements AlbumInfoService {

    @Autowired
    private AlbumInfoMapper albumInfoMapper;

    @Autowired
    private AlbumAttributeValueMapper albumAttributeValueMapper;

    @Autowired
    private KafkaService kafkaService;

    /**
     * 保存专辑
     * 1.保存专辑信息-得到专辑ID
     * 2.保存专辑标签信息
     * 3.保存专辑统计信息
     *
     * @param userId      当前用户ID
     * @param albumInfoVo 专辑VO信息
     */
    @Override
    @Transactional(rollbackFor = Exception.class)  //默认发生RuntimeException事务会进行回滚
    public void saveAlbumInfo(Long userId, AlbumInfoVo albumInfoVo) {
        //1.将提交专辑信息封装AlbumInfo对象，保存1条记录到album_info表
        //1.1 将提交专辑VO对象转为PO对象（持久层对象）
        AlbumInfo albumInfo = BeanUtil.copyProperties(albumInfoVo, AlbumInfo.class);
        //1.2 手动为部分属性进行赋值
        albumInfo.setUserId(userId);
        albumInfo.setTracksForFree(5);
        //todo 刚发布专辑待审状态
        albumInfo.setStatus(SystemConstant.ALBUM_STATUS_PASS);
        //1.3 保存专辑信息得到专辑ID
        //albumInfoMapper.insert(albumInfo);
        this.save(albumInfo);
        //主键回显 通过标签<selectKey>获取  LAST_INSERT_ID()获取记录ID
        Long albumId = albumInfo.getId();

        //2.将提交专辑标签关系封装AlbumAttributeValue集合，批量保存记录到album_attribute_value表
        List<AlbumAttributeValueVo> albumAttributeValueVoList = albumInfoVo.getAlbumAttributeValueVoList();
        if (CollectionUtil.isNotEmpty(albumAttributeValueVoList)) {
            //2.1 将集合类型AlbumAttributeValueVo转为AlbumAttributeValue 关联专辑ID
            for (AlbumAttributeValueVo albumAttributeValueVo : albumAttributeValueVoList) {
                AlbumAttributeValue albumAttributeValue = BeanUtil.copyProperties(albumAttributeValueVo, AlbumAttributeValue.class);
                albumAttributeValue.setAlbumId(albumId);
                //2.2 保存专辑标签关系
                albumAttributeValueMapper.insert(albumAttributeValue);
            }
        }

        //3.批量保存专辑统计信息到album_stat表
        this.saveAlbumStat(albumId, SystemConstant.ALBUM_STAT_PLAY, 0);
        this.saveAlbumStat(albumId, SystemConstant.ALBUM_STAT_SUBSCRIBE, 0);
        this.saveAlbumStat(albumId, SystemConstant.ALBUM_STAT_BUY, 0);
        this.saveAlbumStat(albumId, SystemConstant.ALBUM_STAT_COMMENT, 0);

        //4.TODO 对专辑的内容（文本，图片）进行审核 选择异步

        //5.模拟专辑上下架，通过“是否私密”变量控制上下架
        if ("1".equals(albumInfo.getIsOpen())) {
            kafkaService.sendMessage(KafkaConstant.QUEUE_ALBUM_UPPER, albumId.toString());
        } else {
            kafkaService.sendMessage(KafkaConstant.QUEUE_ALBUM_LOWER, albumId.toString());
        }
    }

    @Autowired
    private AlbumStatMapper albumStatMapper;

    /**
     * 保存专辑统计记录
     *
     * @param albumId  专辑ID
     * @param statType 统计类型
     * @param num      数量
     */
    @Override
    public void saveAlbumStat(Long albumId, String statType, int num) {
        AlbumStat albumStat = new AlbumStat();
        albumStat.setAlbumId(albumId);
        albumStat.setStatType(statType);
        albumStat.setStatNum(num);
        albumStatMapper.insert(albumStat);
    }

    /**
     * 查询当前登录用户发布专辑分页列表(包含统计信息)
     *
     * @param albumInfoQuery 查询条件对象
     * @param pageInfo       分页对象
     * @return
     */
    @Override
    public Page<AlbumListVo> getUserAlbumPage(AlbumInfoQuery albumInfoQuery, Page<AlbumListVo> pageInfo) {
        return albumInfoMapper.getUserAlbumPage(pageInfo, albumInfoQuery);
    }

    @Autowired
    private TrackInfoMapper trackInfoMapper;

    /**
     * 根据专辑ID删除专辑
     *
     * @param id 专辑ID
     * @return
     */
    @Override
    @Transactional(rollbackFor = Exception.class)
    public void removeAlbumInfo(Long id) {
        //1.根据专辑ID查询声音数量 select cout(*) from track_info where album_id = ? limit 1
        LambdaQueryWrapper<TrackInfo> queryWrapper = new LambdaQueryWrapper<>();
        queryWrapper.eq(TrackInfo::getAlbumId, id);
        queryWrapper.last("limit 1");
        Long count = trackInfoMapper.selectCount(queryWrapper);
        if (count > 0) {
            throw new GuiguException(500, "该专辑下已关联声音！");
        }
        //2.删除专辑记录
        albumInfoMapper.deleteById(id);

        //3.删除专辑标签关系记录
        LambdaQueryWrapper<AlbumAttributeValue> albumAttributeValueLambdaQueryWrapper = new LambdaQueryWrapper<>();
        albumAttributeValueLambdaQueryWrapper.eq(AlbumAttributeValue::getAlbumId, id);
        albumAttributeValueMapper.delete(albumAttributeValueLambdaQueryWrapper);

        //4.删除专辑统计记录
        LambdaQueryWrapper<AlbumStat> albumStatLambdaQueryWrapper = new LambdaQueryWrapper<>();
        albumStatLambdaQueryWrapper.eq(AlbumStat::getAlbumId, id);
        albumStatMapper.delete(albumStatLambdaQueryWrapper);

        //5.发送消息删除专辑
        kafkaService.sendMessage(KafkaConstant.QUEUE_ALBUM_LOWER, id.toString());
    }

    @Autowired
    private RedisTemplate redisTemplate;

    /**
     * 根据专辑ID查询专辑信息-包含专辑标签列表
     * 自定义分布式锁
     *
     * @param id
     * @return
     */
    /*@Override
    public AlbumInfo getAlbumInfo(Long id) {
        try {
            //1.优先从缓存中获取数据，命中缓存直接返回，未命中则获取分布式锁
            //1.1 创建缓存业务数据Key 前缀：数据ID标识
            String dataKey = RedisConstant.ALBUM_INFO_PREFIX + id;
            //1.2 查询Redis缓存获取专辑信息
            AlbumInfo albumInfo = (AlbumInfo) redisTemplate.opsForValue().get(dataKey);
            if (albumInfo != null) {
                return albumInfo;
            }
            //2.尝试获取分布式锁
            //2.1 创建锁的Key
            String lockKey = RedisConstant.ALBUM_INFO_PREFIX + id + RedisConstant.CACHE_LOCK_SUFFIX;
            //2.2 创建当前线程锁唯一标识UUID
            String lockValue = IdUtil.randomUUID();
            //2.3 尝试加锁 set k v ex nx
            Boolean flag = redisTemplate.opsForValue().setIfAbsent(lockKey, lockValue, 5, TimeUnit.SECONDS);

            //3.获取分布式锁成功：执行业务代码
            if (flag) {
                //3.1 由于前置代码中已有分布式布隆过滤器对查询专辑进行判断是否存在
                try {
                    AlbumInfo albumInfoFromDB = this.getAlbumInfoFromDB(id);
                    long ttl = RedisConstant.ALBUM_TIMEOUT + RandomUtil.randomInt(0, 500);
                    redisTemplate.opsForValue().set(dataKey, albumInfoFromDB, ttl, TimeUnit.SECONDS);
                    return albumInfoFromDB;
                } finally {
                    //4.业务执行完毕，释放锁 采用Lua脚本释放锁
                    String scriptText = "if redis.call(\"get\",KEYS[1]) == ARGV[1]\n" +
                            "then\n" +
                            "    return redis.call(\"del\",KEYS[1])\n" +
                            "else\n" +
                            "    return 0\n" +
                            "end";
                    //创建脚本对象-封装lua脚本
                    DefaultRedisScript<Long> redisScript = new DefaultRedisScript<>();
                    redisScript.setScriptText(scriptText);
                    redisScript.setResultType(Long.class);
                    redisTemplate.execute(redisScript, Arrays.asList(lockKey), lockValue);
                }
            } else {
                //5.获取分布式锁失败：自旋
                Thread.sleep(100);
                return this.getAlbumInfo(id);
            }
        } catch (Exception e) {
            log.error("[专辑服务]查询专辑信息异常兜底处理：{}", e);
            //兜底处理方案：当前Redis服务不可用（缓存，分布式锁），直接查询数据库
            return this.getAlbumInfoFromDB(id);
        }
    }*/

    @Autowired
    private RedissonClient redissonClient;

    /**
     * 根据专辑ID查询专辑信息-包含专辑标签列表
     * Redisson分布式锁
     *
     * @param id 1(100线程) 2（100线程）
     * @return
     */
    @Override
    public AlbumInfo getAlbumInfo(Long id) {
        try {
            //1.优先从缓存中获取数据，命中返回直接返回，未命中获取分布式锁
            //1.1 创建缓存业务数据Key 形式：前缀+数据ID标识
            String dataKey = RedisConstant.ALBUM_INFO_PREFIX + id;
            //1.2 查询缓存中专辑业务数据
            AlbumInfo albumInfo = (AlbumInfo) redisTemplate.opsForValue().get(dataKey);
            if (albumInfo != null) {
                return albumInfo;
            }
            //2.获取分布式锁-最终所有来到该位置线程终究会获取锁成功
            //2.1 创建锁Key 形式：前缀+数据标识+锁后缀
            String lockKey = RedisConstant.ALBUM_INFO_PREFIX + id + RedisConstant.CACHE_LOCK_SUFFIX;
            //2.2 创建锁对象
            RLock lock = redissonClient.getLock(lockKey);
            //2.3 执行加锁方法
            lock.lock();

            //3.执行业务（查询数据库业务）
            try {
                //3.1 只要线程调用lock方法最终都会获取锁成功，避免后来线程获取锁成功造成多次查库，再查一次缓存
                albumInfo = (AlbumInfo) redisTemplate.opsForValue().get(dataKey);
                if (albumInfo != null) {
                    return albumInfo;
                }
                //3.2 执行查询数据库方法 将查询结果放入缓存
                albumInfo = this.getAlbumInfoFromDB(id);
                long ttl = RedisConstant.ALBUM_TIMEOUT + RandomUtil.randomInt(0, 500);
                redisTemplate.opsForValue().set(dataKey, albumInfo, ttl, TimeUnit.SECONDS);
                return albumInfo;
            } finally {
                //4.释放锁
                lock.unlock();
            }
        } catch (Exception e) {
            log.error("[专辑服务]查询专辑信息异常兜底处理：{}", e);
            //兜底处理方案：当前Redis服务不可用（缓存，分布式锁），直接查询数据库
            return this.getAlbumInfoFromDB(id);
        }
    }


    /**
     * 查询专辑统计信息(从数据中查询获取)
     *
     * @param id
     * @return
     */
    @Override
    public AlbumInfo getAlbumInfoFromDB(Long id) {
        //1.根据专辑ID查询专辑信息
        AlbumInfo albumInfo = albumInfoMapper.selectById(id);

        //2.根据专辑ID查询标签关系列表
        LambdaQueryWrapper<AlbumAttributeValue> queryWrapper = new LambdaQueryWrapper<>();
        queryWrapper.eq(AlbumAttributeValue::getAlbumId, id);
        List<AlbumAttributeValue> albumAttributeValueList = albumAttributeValueMapper.selectList(queryWrapper);
        if (CollectionUtil.isNotEmpty(albumAttributeValueList)) {
            albumInfo.setAlbumAttributeValueVoList(albumAttributeValueList);
        }
        return albumInfo;
    }

    /**
     * 更新专辑信息
     *
     * @param albumInfo
     */
    @Override
    @Transactional(rollbackFor = Exception.class)
    public void updateAlbumInfo(AlbumInfo albumInfo) {
        //1.修改专辑表信息
        this.updateById(albumInfo);

        //2.修改专辑标签关系记录
        //2.1 根据专辑ID删除“旧”的专辑标签记录
        LambdaQueryWrapper<AlbumAttributeValue> albumAttributeValueLambdaQueryWrapper = new LambdaQueryWrapper<>();
        albumAttributeValueLambdaQueryWrapper.eq(AlbumAttributeValue::getAlbumId, albumInfo.getId());
        albumAttributeValueMapper.delete(albumAttributeValueLambdaQueryWrapper);

        //2.2 新增专辑标签关系记录
        List<AlbumAttributeValue> albumAttributeValueVoList = albumInfo.getAlbumAttributeValueVoList();
        if (CollectionUtil.isNotEmpty(albumAttributeValueVoList)) {
            //2.1 将集合类型AlbumAttributeValueVo转为AlbumAttributeValue 关联专辑ID
            for (AlbumAttributeValue albumAttributeValue : albumAttributeValueVoList) {
                albumAttributeValue.setAlbumId(albumInfo.getId());
                //2.2 保存专辑标签关系
                albumAttributeValueMapper.insert(albumAttributeValue);
            }
        }
        //3.模拟专辑上下架，通过“是否私密”变量控制上下架
        if ("1".equals(albumInfo.getIsOpen())) {
            kafkaService.sendMessage(KafkaConstant.QUEUE_ALBUM_UPPER, albumInfo.getId().toString());
        } else {
            kafkaService.sendMessage(KafkaConstant.QUEUE_ALBUM_LOWER, albumInfo.getId().toString());
        }
    }

    /**
     * 查询当前用户发布专辑列表
     *
     * @param userId
     * @return
     */
    @Override
    public List<AlbumInfo> getUserAllAlbumList(Long userId) {
        //select id,album_title from album_info where user_id = ? and is_deleted = 0;
        LambdaQueryWrapper<AlbumInfo> queryWrapper = new LambdaQueryWrapper<>();
        //1.查询条件
        queryWrapper.eq(AlbumInfo::getUserId, userId);
        queryWrapper.eq(AlbumInfo::getStatus, SystemConstant.ALBUM_STATUS_PASS);
        //2.指定查询列
        queryWrapper.select(AlbumInfo::getId, AlbumInfo::getAlbumTitle);
        //3.TODO 限制返回记录数 太多导致小程序卡顿
        queryWrapper.last("limit 100");
        //4.排序
        queryWrapper.orderByDesc(AlbumInfo::getId);
        return albumInfoMapper.selectList(queryWrapper);
    }

    /**
     * 将指定的专辑发布
     *
     * @param id
     * @return
     */
    @Override
    public void publishAlbum(Long id) {
        //1.查询专辑下所有声音是否都经过审核
        LambdaQueryWrapper<TrackInfo> queryWrapper = new LambdaQueryWrapper<>();
        queryWrapper.eq(TrackInfo::getStatus, SystemConstant.TRACK_STATUS_REVIEWING).or().eq(TrackInfo::getStatus, SystemConstant.TRACK_STATUS_NO_PASS);
        queryWrapper.eq(TrackInfo::getAlbumId, id);
        queryWrapper.last("limit 1");
        queryWrapper.select(TrackInfo::getId);
        Long count = trackInfoMapper.selectCount(queryWrapper);
        if (count > 0) {
            throw new GuiguException(500, "该专辑下存在未审核通过声音");
        }
        //2.发送MQ消息到Kafka 异步 上架专辑
        kafkaService.sendMessage(KafkaConstant.QUEUE_ALBUM_UPPER, id.toString());
    }

    /**
     * 查询专辑统计信息
     *
     * @param albumId
     * @return
     */
    @Override
    @GuiGuCache(prefix=RedisConstant.ALBUM_INFO_PREFIX+"stat:")
    public AlbumStatVo getAlbumStatVo(Long albumId) {
        return albumInfoMapper.getAlbumStatVo(albumId);
    }


}
